/*
 * Middleware
 * Copyright (C) 2010/11
 * Riccardo Pulvirenti, Giuseppe Ravidà, Andrea Tino
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANDABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see http://www.gnu.org/licenses
 */

/*!
 * File: worker.cpp
 */

#include "utils.hpp"
#include "worker.hpp"
#include "discovery_proxy_factory.hpp"
#include "data_proxy_factory.hpp"
#include "data.hpp"

using namespace middleware;
using namespace middleware::logging;

//-----------------------------------------------------------------------------
// Constructors and destructor
/* Default constructor */
Worker::Worker(PeersKnowledge PK, WorkerDescriptor descriptor) {
	this->_worker_descriptor = descriptor;
	this->_pk = PK;
	// Xml logger
	this->_config._xml_logger = 0;
}
/* With logger constructor */
Worker::Worker(WConfig conf, PeersKnowledge PK, WorkerDescriptor descriptor) {
	this->_worker_descriptor = descriptor;
	this->_pk = PK;
	// Xml logger
	this->_config._xml_logger = conf._xml_logger;
}
/* Destructor */
Worker::~Worker() {

}
//-----------------------------------------------------------------------------
// Private members
/* Thread execution flow */
void Worker::exec() {
	// Execution flow for Worker
	std::cout << kWorkerEtq << "Running!" << std::endl;
	this->bonjour(); /* Initialization sequence */
	try {
		// Launching thread to manage status
		std::cout << kWorkerEtq <<
				"Launching worker status manager thread..." << std::endl;
		this->_workerstatus_manager_thread = boost::shared_ptr<boost::thread>(
				new boost::thread(boost::bind(
						&Worker::exec_workerstatus_manager,	this)));
		// Launching thread to manage incoming task data
		std::cout << kWorkerEtq <<
				"Launching task data manager thread..." << std::endl;
		this->_taskdata_manager_thread = boost::shared_ptr<boost::thread>(
				new boost::thread(boost::bind(
						&Worker::exec_taskdata_manager,	this)));
		// Launching thread to collect TDC
		std::cout << kWorkerEtq <<
				"Launching TDC Collector thread..." << std::endl;
		this->_taskdata_collector_thread = boost::shared_ptr<boost::thread>(
				new boost::thread(boost::bind(
						&Worker::exec_taskdata_collector, this)));
		// Cycling and dequeueing from queue
		for (;;) {
			std::cout << kWorkerEtq << "Dequeueing loop, " <<
					"Dequeueing task..." << std::endl;
			TaskDescriptor td = this->_tqueue_manager.dequeue(); /* Blocking */
			std::cout << kWorkerEtq << "Dequeued task id=" << td.get_id() <<
					". TQueue has now " << this->_tqueue_manager.size() <<
					" tasks." << " Now managing dequeued task..." << std::endl;
			*(this->_config._xml_logger) << TaskDequeued(&td); /* LOG */
			// A thread will manage the task (multithread)
			boost::shared_ptr<boost::thread>(new boost::thread(
					boost::bind(&Worker::ManageTask, this, td)));
			//this->ManageTask(td); /* Managing this flow (singlethread) */
			// Update status
			boost::mutex::scoped_lock lck(this->_worker_descriptor_mutex);
			WorkerLoad status = this->_worker_descriptor.get_status();
			this->_worker_descriptor.set_status(status - 1);
			lck.unlock();
			std::cout << kWorkerEtq << "Status changed: Current Load = " <<
					(status - 1) << std::endl;
		}
	} catch(std::exception& e) {
		std::cout << kWorkerEtq << "Error: " << e.what() << std::endl;
		std::cout << kWorkerEtq << "Now exiting!" << std::endl;
	}
}
/* Function enqueuer called through proxy by Manager | Sharing a SynQueue */
void Worker::enqueuer(const TaskDescriptor& td) {
	using boost::asio::ip::tcp;
	std::cout << kWorkerEtq << "Task enqueuer called!" << std::endl;
	try {
		std::cout << kWorkerEtq << "Task descriptor to enqueue: TID=" <<
				td.get_id() << ". Now enqueueing..."  << std::endl;
		// Now enqueueing the task
		this->_tqueue_manager.enqueue(td);
		std::cout << kWorkerEtq << "Task descriptor " << td.get_id() <<
				" enqueued!" << std::endl;
		{ /* LOG */
			TaskDescriptor tmp_td = td;
			*(this->_config._xml_logger) << TaskEnqueued(&tmp_td, "MAN");
		}
		// Update status
		boost::mutex::scoped_lock lck(this->_worker_descriptor_mutex);
		WorkerLoad status = this->_worker_descriptor.get_status();
		this->_worker_descriptor.set_status(status + 1);
		lck.unlock();
		std::cout << kWorkerEtq << "Status changed: Current Load = " <<
				(status + 1) << std::endl;
	} catch(std::exception& e) {
		std::cout << kWorkerEtq << "Error: " << e.what() << std::endl;
		std::cout << kWorkerEtq << "Program now dying..." << std::endl;
		exit(1);
	}
}
/* Sending periodically worker status to discovery */
void Worker::exec_workerstatus_manager() {
	/* Activate for a certain period of time and, when self-triggered, check
	 * for some changes to have happened in worker status, if so, collect
	 * info and send them to discovery */
	try {
		std::cout << kWorkerEtq << kWorker1Etq << "Running!" << std::endl;
		bool init = true; /* First time the control loop is executed? */
		WorkerDescriptor prev_wdesc; /* To check diffs with previous status */
		std::cout << kWorkerEtq << kWorker1Etq <<
				"Entering worker status manager loop..." << std::endl;
		for (;;) {
			std::cout << kWorkerEtq << kWorker1Etq << "Reactivated..." <<
					std::endl;
			// Logging status notification (changed or not?)
			{ /* LOG TODO strange values*/
				bool changed = ((this->_worker_descriptor != prev_wdesc));
				*(this->_config._xml_logger) << StatusUpdateLoop(changed,
						&prev_wdesc, &(this->_worker_descriptor));
			}
#ifdef SENDWORKERDESC_ON_STATUS_CHANGE
			if ((this->_worker_descriptor != prev_wdesc) || init) {
#else
			if (true) { /* Always notify */
#endif
				/* If this is the first time the control loop is executed or
				 * a change has been detected, we must send this worker's
				 * status */
				std::cout << kWorkerEtq << kWorker1Etq <<
						"Updating status to peers..." <<
						std::endl;
#ifdef SENDWORKERDESC_ON_STATUS_CHANGE
				*(this->_config._xml_logger) << Event("W", std::string(
						"Periodic status manager loop detected changes to notify.")); /* LOG */
#else
				*(this->_config._xml_logger) << Event("W", std::string(
						"Periodic status manager loop now notifying.")); /* LOG */

#endif
				/* Locking to enter in critical region for PeersKnowledge */
				boost::mutex::scoped_lock lck(
						*(this->_pk.ptr_to_worker_descriptors_mutex));
				PeersSet::iterator it;
				// We must browse all worker descriptors
				for (it = this->_pk.ptr_to_worker_descriptors->begin();
						it < this->_pk.ptr_to_worker_descriptors->end();
						it++) {
					// Create factory, get proxy and update worker descriptor
					DiscoveryProxyFactory proxy_factory(
							it->get_address().get_ip(),
							boost::lexical_cast<std::string>(
									it->get_address().get_port_disc()));
					WorkerDiscoveryProxy* proxy = proxy_factory.getProxy();
					try {
						proxy->update_worker(this->_worker_descriptor);
						std::cout << kWorkerEtq << kWorker1Etq <<
								"Worker descriptor change notified to " <<
								"peer at IP=" <<
								it->get_address().get_ip() << ", PORT=" <<
								it->get_address().get_port_disc() <<
								" (disc), new PI=" <<
								this->_worker_descriptor.PI() << std::endl;
						{  /* LOG */
							WorkerDescriptor tmp_wd = (*it);
							Address tmp_addr = it->get_address();
							*(this->_config._xml_logger) <<
									WorkerDescriptorSent(&tmp_wd, &tmp_addr);
						}
					} catch(ProxyException& e) {
						std::cout << kWorkerEtq << kWorker1Etq <<
								"Worker descriptor change could not " <<
								"be notified for a network problem: " <<
								e.what() << std::endl;
						*(this->_config._xml_logger) << Error(std::string(
								"Periodic status manager loop COULD NOT notify ") +
								std::string("this worker descriptor to ") +
								std::string("peer at IP=") +
								it->get_address().get_ip() +
								std::string(" on PORT=") +
								boost::lexical_cast<std::string>(
										it->get_address().get_port_disc()) +
										std::string(" (disc); new PI=") +
										boost::lexical_cast<std::string>(
												this->_worker_descriptor.PI()) +
												std::string("."), "W"); /* LOG */
					}
					/*std::cout << kWorkerEtq << kWorker1Etq <<
							"WorkerDesc " << it->PI() << std::endl;*/
				}
				lck.unlock(); /* Unlock */
				prev_wdesc = this->_worker_descriptor; /* Update */
				init = false;
			} else {
				// No changes
				std::cout << kWorkerEtq << kWorker1Etq <<
						"No Changes detected!" <<
						" Going to sleep and reactivate later..." << std::endl;
			}
			// Waiting and reactivating then again
			boost::this_thread::sleep(boost::posix_time::milliseconds(
					kWorkerStatusManagerRefreshPeriodMsec));
		}
	} catch(std::exception& e) {
		/* Proxy reference is automatically destroys as the factory gets
		 * out of scope */
		std::cout << kWorkerEtq << kWorker1Etq << "Error: " << e.what() <<
				std::endl;
		std::cout << kWorkerEtq << kWorker1Etq << "Now exiting!" << std::endl;
	}
}
/* Data manager */
void Worker::exec_taskdata_manager() {
	std::cout << kWorkerEtq << kWorker2Etq << "Running!" << std::endl;
	try {
		std::cout << kWorkerEtq << kWorker2Etq << "Setting factory..." << std::endl;
		DataProxyFactory proxy_factory(kReceiveDataProxy,
				this->_worker_descriptor.get_address().get_ip(),
				boost::lexical_cast<std::string>(
						this->_worker_descriptor.get_address().get_port_w()));
		std::cout << kWorkerEtq << kWorker2Etq <<
				"Getting proxy binded on IP=" <<
				this->_worker_descriptor.get_address().get_ip() <<
				" and PORT=" <<
				this->_worker_descriptor.get_address().get_port_w() << std::endl;
		DataManagerProxy* proxy = proxy_factory.getProxy();
		std::cout << kWorkerEtq << kWorker2Etq <<
				"Entering data receive loop..." << std::endl;
		for (;;) {
			std::cout << kWorkerEtq << kWorker2Etq <<
					"Waiting for task data..." << std::endl;
			Data data = proxy->recv_data();
			std::cout << kWorkerEtq << kWorker2Etq <<
					"ARRIVED TASK DATA! Data of task TID=" <<
					data.get_owner_task_id() << ", data is: " <<
					data.get_data_str() << "/" << data.get_data_int() <<
					". Now inserting in Task Data Collection (TDC)" << std::endl;
			*(this->_config._xml_logger) << Event("W", std::string(
					"Arrived task data for task TID=") +
					boost::lexical_cast<std::string>(data.get_owner_task_id()) +
					std::string(". Data is: ") +
					data.get_data_str() + std::string("/") +
					boost::lexical_cast<std::string>(data.get_data_int()) +
					std::string(".")); /* LOG */
			// Insert in TDC
			/* Critical region */
			boost::mutex::scoped_lock lck(this->_taskdata_collection_mutex);
			TDCEntry tdce = {kWorkerTDCEntryTTL, data};
			this->_taskdata_collection.push_back(tdce);
			std::cout << kWorkerEtq << kWorker2Etq <<
					"Data of task TID=" << data.get_owner_task_id() <<
					" has been put in TDC! TDC has now " <<
					this->_taskdata_collection.size() << " task data" << std::endl;
			lck.unlock(); /* Unlocking */
			*(this->_config._xml_logger) << Event("W", std::string(
					"Task data for task TID=") +
					boost::lexical_cast<std::string>(data.get_owner_task_id()) +
					std::string(", data was: ") +
					data.get_data_str() + std::string("/") +
					boost::lexical_cast<std::string>(data.get_data_int()) +
					std::string(", has been added to TDC.")); /* LOG */
		}
	} catch(std::exception& e) {
		std::cout << kWorkerEtq << kWorker2Etq << "Error: " << e.what() <<
				std::endl;
		std::cout << kWorkerEtq << kWorker2Etq << "Now exiting!" << std::endl;
	}
}
/* Managing a task (can be a thread) */
void Worker::ManageTask(TaskDescriptor task) {
	std::cout << kWorkerEtq << "TManager: " << "Managing task "
			<< task.get_id() << ", type = " << task.get_type() << std::endl;
	if (task.get_type() == kTaskBashCommand) {
		// If this is just a command and not an executable, do this and get away
		std::cout << kWorkerEtq << "TManager: " << "Command Task " <<
				task.get_id() << " ***EXECUTED***!" << std::endl;
		*(this->_config._xml_logger) << TaskExecuted(&task); /* LOG */
		return;
	}
	// If this is an executable, retrieve data
	std::cout << kWorkerEtq << "TManager: " << " Task TID=" <<
			task.get_id() << " is an executable, need to find data..." << std::endl;
	/* Data might not be arrived yet, let's give them the time */
	for (int j = kWorkerGetDataAttemptsNumber; true; j--) {
		// Check
		if (j <= 0) {
			// Sorry, no more try... data is missing
			std::cout << kWorkerEtq << "TManager: " << "Executable Task " <<
					task.get_id() <<
					" ***DEFINETELY NOT EXECUTED***! data not found!" <<
					std::endl;
			*(this->_config._xml_logger) << Error(std::string(
					"Command task TID=") +
					boost::lexical_cast<std::string>(task.get_id()) +
					std::string(" ***DEFINETELY NOT EXECUTED***. Data not found."),
					"W"); /* LOG */
			return;
		}
		/* Critical region */
		boost::mutex::scoped_lock lck(this->_taskdata_collection_mutex);
		TaskDataCollection::iterator it;
		Data task_data;
		bool found = false;
		for (it = this->_taskdata_collection.begin();
				it < this->_taskdata_collection.end(); it++) {
			if (it->data.get_owner_task_id() == task.get_id()) {
				// Found data: take it and erase from the collection
				task_data = it->data;
				this->_taskdata_collection.erase(it);
				found = true;
				std::cout << kWorkerEtq << "TManager: " << "TDC has now " <<
						this->_taskdata_collection.size() << " task data" << std::endl;
				break;
			}
		}
		lck.unlock(); /* Unlock */
		if (found) {
			std::cout << kWorkerEtq << "TManager: " << "Executable Task " <<
					task.get_id() << " ***EXECUTED***! with data: " <<
					task_data.get_data_str() << "/" <<
					task_data.get_data_int() << std::endl;
			*(this->_config._xml_logger) << TaskExecuted(&task); /* LOG */
			return;
		} else {
			std::cout << kWorkerEtq << "TManager: " << "Executable Task " <<
					task.get_id() << " ***STILL NOT EXECUTED***! data not yet arrived!" <<
					" Maybe data must still arrive... trying later..." << std::endl;
			*(this->_config._xml_logger) << ExecTaskStillNotExecuted(&task,
					(unsigned int)(kWorkerGetDataAttemptsNumber - j + 1),
					(unsigned int)kWorkerGetDataAttemptsNumber); /* LOG */
			boost::this_thread::sleep(boost::posix_time::milliseconds(
					kWorkerGetDataWaitTimeMsec));
		}
	}
}
/* TDC Collector */
void Worker::exec_taskdata_collector() {
	std::cout << kWorkerEtq << kWorker3Etq << "Running!" << std::endl;
	// Cycling
	for (;;) {
		std::cout << kWorkerEtq << kWorker3Etq <<
				"TDC Collection loop reactivated..." <<	std::endl;
		unsigned int collected = 0;
		unsigned int current_size = 0;
		// Locking
		boost::mutex::scoped_lock lck(this->_taskdata_collection_mutex);
		TaskDataCollection::iterator it;
		for (it = this->_taskdata_collection.begin();
				it < this->_taskdata_collection.end();) {
			if (it->ttl-- <= 0) {
				// Need to collect
				it = this->_taskdata_collection.erase(it); /* Also update next */
				collected++;
			} else {
				// No need to collect
				it++; /* Incrementing */
			}
		}
		current_size = this->_taskdata_collection.size();
		// Unlocking
		lck.unlock();
		// Notifying
		if (collected > 0) {
			std::cout << kWorkerEtq << kWorker3Etq << "TDCCollector: " <<
					collected <<
					" entries were COLLECTED! Current TDC size: " <<
					current_size <<	" entries." << std::endl;
		} else {
			std::cout << kWorkerEtq << kWorker3Etq << "TDCCollector: " <<
					"Nothing collected! Current TDC size: " << current_size <<
					" entries." << std::endl;
		}
		*(this->_config._xml_logger) << TDCCollectorLoopExec(collected); /* LOG */
		// Waiting
		boost::this_thread::sleep(boost::posix_time::milliseconds(
				kWorkerTDCCollectionPeriodMsec));
	}
}
/* Bonjour */
void Worker::bonjour() {
	std::cout << kWorkerEtq << "Initialization sequence now started..." <<
				std::endl;
	std::stringstream ss;
	ss << "Configuration loaded: GateThroughMan_IP: this=" <<
			this->_worker_descriptor.get_address().get_ip()  <<
			", GateTrhoughMan_PORT: " <<
			this->_worker_descriptor.get_address().get_port() <<
			" | bind_data_IP=" <<
			this->_worker_descriptor.get_address().get_ip() <<
			", bind_data_PORT=" <<
			this->_worker_descriptor.get_address().get_port_w() << " *** " <<
			"Worker capabilities: " <<
			"CPU: " <<
			this->_worker_descriptor._capabilities.capab_cpu_mhz <<
			" MHz; " <<	"RAM: " <<
			this->_worker_descriptor._capabilities.capab_ram_mb << " MB; " <<
			"Cores: " <<
			this->_worker_descriptor._capabilities.capab_core_num <<
			" cores; " << "Processors: " <<
			this->_worker_descriptor._capabilities.capab_cpu_num <<
			" cpus; " << "Distance: " <<
			this->_worker_descriptor._capabilities.capab_hops <<
			" hops; " << "Bandwidth: " <<
			this->_worker_descriptor._capabilities.capab_bandwidth <<
			" Kbit/s.";
	std::cout << kWorkerEtq << ss.str() << std::endl;
	std::cout << kWorkerEtq << "Initialization completed!" << std::endl;
	std::cout << kWorkerEtq << "Report ended!" << std::endl;
	*(this->_config._xml_logger) << Event("W", ss.str());
}
